package org.apache.rocketmq.remoting.netty;

import com.google.common.base.Stopwatch;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import io.netty.util.TimerTask;
import org.apache.rocketmq.remoting.protocol.ResponseCode;

/**
 * @author: jiaxiangen
 * @Date: 2025/6/11 10:29
 * @Version 1.0
 * @Description: netty 构建的客户端
 */
public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {


    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);


    //获取同步锁的超时时间
    private static final long LOCK_TIMEOUT_MILLIS = 3000;

    //关闭channel的最小超时时间
    private static final long MIN_CLOSE_TIMEOUT_MILLIS = 100;
    //下面这些都是构建Netty服务器要用到的成员变量，大家应该很熟悉了吧？
    //Netty服务器的启动器
    private final Bootstrap bootstrap = new Bootstrap();

    //处理SocketChannel的IO事件的循环组
    private final EventLoopGroup eventLoopGroupWorker;
    //Netty客户端配置
    private final NettyClientConfig nettyClientConfig;

    private final Lock lockChannelTables = new ReentrantLock();
    private final NettyEventExecutor nettyEventExecutor = new NettyEventExecutor();

    //存储服务器网络地址和对应的channel包装对象键值对的成员变量
    private final ConcurrentHashMap<String, ChannelWrapper> channelTables = new ConcurrentHashMap<>();

    //存储channel和对应的channel包装对象键值对的成员变量
    private final ConcurrentHashMap<Channel, ChannelWrapper> channelWrapperTables = new ConcurrentHashMap<>();

    //时间轮定时任务调度器
    private final HashedWheelTimer timer = new HashedWheelTimer(r -> new Thread(r, "ClientHouseKeepingService"));

    //存储Namesrv地址列表的原子引用对象
    private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<>();


    //存储可用的namesrv地址和对应的是否可用的状态
    private final ConcurrentHashMap<String, Boolean> availableNamesrvAddrMap = new ConcurrentHashMap<>();

    //Namesrv可用地址选择器
    private final AtomicReference<String> namesrvAddrChoosed = new AtomicReference<>();

    //保证通信正常获取
    private final Lock namesrvChannelLock = new ReentrantLock();

    //计算要使用的Namesrv地址的索引
    private final AtomicInteger namesrvAddrIndex = new AtomicInteger(initValueIndex());
    //同步锁，用来保证创建与Namesrv通信的channel的并发安全
    private final Lock lockChannelAddr = new ReentrantLock();

    //公共执行器，用于执行客户端收到响应后的回调方法
    private final ExecutorService publicExecutor;
    //这个执行器用于定义更新与Namesrv通信的channel
    private final ExecutorService scanExecutor;

    //回调方法执行器，在Netty构建的服务端，这个执行器会用来执行收到响应后的回调方法
    //在为Broker模块构建的Netty客户端，并没有给这个成员变量赋值，而是使用了publicExecutor来执行收到响应后的回调方法
    private ExecutorService callbackExecutor;


    //默认事件循环组
    private EventExecutorGroup defaultEventExecutorGroup;
    //在Netty构建的服务端给这个成员变量赋值了，在为Broker模块构建的Netty客户端中并没有给这个成员变量赋值
    private final ChannelEventListener channelEventListener;


    public NettyRemotingClient(final NettyClientConfig nettyClientConfig) {
        this(nettyClientConfig, null);
    }


    //构造方法
    public NettyRemotingClient(final NettyClientConfig nettyClientConfig,
                               final ChannelEventListener channelEventListener) {
        this(nettyClientConfig, channelEventListener, null, null);
    }


    //真正创建NettyRemotingClient客户端对象的构造方法
    public NettyRemotingClient(final NettyClientConfig nettyClientConfig, final ChannelEventListener channelEventListener, final EventLoopGroup eventLoopGroup, final EventExecutorGroup eventExecutorGroup) {
        //这里的操作在Netty构建的服务端中已经见过了，设置单向信号量值和异步请求信号量值
        super(nettyClientConfig.getClientOnewaySemaphoreValue(), nettyClientConfig.getClientAsyncSemaphoreValue());
        this.nettyClientConfig = nettyClientConfig;
        this.channelEventListener = channelEventListener;
        //从客户端中得到信息
        int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            //获取当前计算机核数
            publicThreadNums = Runtime.getRuntime().availableProcessors();
        }
        this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactoryImpl("NettyClientPublicExecutor_"));
        //创建扫描执行器
        this.scanExecutor = ThreadUtils.newThreadPoolExecutor(4, 10, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(32), new ThreadFactoryImpl("NettyClient_ScanResponseTable_"));
        //创建默认的EventLoopGroup
        if (eventLoopGroup == null) {
            this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactoryImpl("NettyClient_"));
        } else {
            this.eventLoopGroupWorker = eventLoopGroup;
        }
        //创建EventLoopGroup
        this.defaultEventExecutorGroup = eventExecutorGroup;
    }

    private int initValueIndex() {
        return (int) (System.currentTimeMillis() % 999);
    }

    //@方法描述：启动Broker的客户端的方法

    @Override
    public void start() {
        if (this.defaultEventExecutorGroup == null) {
            this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(nettyClientConfig.getClientWorkerThreads(), new ThreadFactoryImpl("NettyClientWorkerThread_"));
        }
        Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(
                                nettyClientConfig.isDisableNettyWorkerGroup() ? null : defaultEventExecutorGroup,
                                new NettyEncoder(),
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                                new NettyConnectManageHandler(),
                                new NettyClientHandler());
                    }
                });
        //从客户端配置信息对象中得到配置的缓冲区大小，设置客户端发送缓冲区的大小
        if (nettyClientConfig.getClientSocketSndBufSize() > 0) {
            LOGGER.info("client set SO_SNDBUF to {}", nettyClientConfig.getClientSocketSndBufSize());
            handler.option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize());
        }
        //从客户端配置信息对象中得到配置的缓冲区大小，设置客户端接收缓冲区的大小
        if (nettyClientConfig.getClientSocketRcvBufSize() > 0) {
            LOGGER.info("client set SO_RCVBUF to {}", nettyClientConfig.getClientSocketRcvBufSize());
            handler.option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize());
        }

        //设置客户端写缓冲区高低水位线
        if (nettyClientConfig.getWriteBufferLowWaterMark() > 0 && nettyClientConfig.getWriteBufferHighWaterMark() > 0) {
            LOGGER.info("client set netty WRITE_BUFFER_WATER_MARK to {},{}",
                    nettyClientConfig.getWriteBufferLowWaterMark(), nettyClientConfig.getWriteBufferHighWaterMark());
            handler.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                    nettyClientConfig.getWriteBufferLowWaterMark(), nettyClientConfig.getWriteBufferHighWaterMark()));
        }
        //根据配置信息判断客户端是否启用了池化的内存分配功能
        if (nettyClientConfig.isClientPooledByteBufAllocatorEnable()) {
            //如果启用了则设置池化内存分配器
            handler.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }
        //启动netty事件执行器
        nettyEventExecutor.start();

        TimerTask scanResponseTableTask = new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                try {
                    NettyRemotingClient.this.scanResponseTable();
                } catch (Throwable e) {
                    LOGGER.error("scanResponseTable exception", e);
                } finally {
                    //在任务执行完毕之后，再次把当前任务提交给时间轮
                    timer.newTimeout(this, 1000, TimeUnit.MILLISECONDS);
                }
            }
        };
        //把刚才创建的扫描客户端响应表的任务提交给时间轮，在3秒之后执行该任务
        timer.newTimeout(scanResponseTableTask, 1000 * 3, TimeUnit.MILLISECONDS);

        //配置信息中得到连接超时时间
        int connectTimeoutMillis = this.nettyClientConfig.getConnectTimeoutMillis();
        //定义一个定时更新availableNamesrvAddrMap成员变量的任务，其实就是定时更新可用的Namesrv地址的任务
        TimerTask timerTaskScanAvailableNameSrv = new TimerTask() {
            @Override
            public void run(Timeout timeout) {
                try {
                    //执行扫描可用的Namesrv地址的方法
                    NettyRemotingClient.this.scanAvailableNameSrv();
                } catch (Exception e) {
                    LOGGER.error("scanAvailableNameSrv exception", e);
                } finally {
                    //在任务执行完毕之后，再次把当前任务提交给时间轮，下次执行时间为连接超时时间
                    timer.newTimeout(this, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                }
            }
        };
        //把刚才创建好的定时任务提交给时间轮，这里没有延迟，立即就会执行一次timerTaskScanAvailableNameSrv任务
        //也就是说，随着Broker的启动，Broker模块Netty客户端的创建，就会立即执行一次更新可用的Namesrv地址的任务
        //把可用的Namesrv的地址设置到availableNamesrvAddrMap成员变量中
        this.timer.newTimeout(timerTaskScanAvailableNameSrv, 0, TimeUnit.MILLISECONDS);


    }

    private void scanAvailableNameSrv() {
        List<String> namesrvList = this.namesrvAddrList.get();
        if (namesrvList == null) {
            LOGGER.info("begin scan available name server address");
            return;
        }
        //判断当前不在扫描可用的Namesrv地址的状态
        for (Map.Entry<String, Boolean> entry : this.availableNamesrvAddrMap.entrySet()) {
            String availableNameAddr = entry.getKey();
            if (!namesrvList.contains(availableNameAddr)) {
                //如果当前可可用的Namesrv地址不在可可用的Namesrv地址集合中，则从可可用的Namesrv地址集合中删除该地址
                LOGGER.info("remove available name server address {}", availableNameAddr);
                this.availableNamesrvAddrMap.remove(availableNameAddr);
            }
        }
        for (String nameAddr : namesrvList) {
            //创建通道
            scanExecutor.execute(() -> {
                try {
                    Channel channel = this.getAndCreateChannel(nameAddr);
                    if (channel != null && channel.isActive()) {
                        //如果通道是活跃的，则将可可用的Namesrv地址添加到可可用的Namesrv地址集合中
                        this.availableNamesrvAddrMap.put(nameAddr, true);
                    } else {
                        //如果通道不是活跃的，则从可可用的Namesrv地址集合中删除该地址
                        this.availableNamesrvAddrMap.remove(nameAddr);
                    }

                } catch (Exception e) {
                    LOGGER.error("scan available name server address exception", e);
                }
            });
        }

    }

    @Override
    public void shutdown() {
        try {
            this.timer.stop();
            for (String addr : this.channelTables.keySet()) {
                this.channelTables.get(addr).close();
            }
            this.channelWrapperTables.clear();
            this.channelTables.clear();
            this.eventLoopGroupWorker.shutdownGracefully();
            if (this.nettyEventExecutor != null) {
                this.nettyEventExecutor.shutdown();
            }
            if (this.defaultEventExecutorGroup != null) {
                this.defaultEventExecutorGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            LOGGER.error("NettyRemotingClient shutdown exception, ", e);
        }
        if (this.publicExecutor != null) {
            try {
                this.publicExecutor.shutdown();
            } catch (Exception e) {
                LOGGER.error("NettyRemotingServer shutdown exception, ", e);
            }
        }
        if (this.scanExecutor != null) {
            try {
                this.scanExecutor.shutdown();
            } catch (Exception e) {
                LOGGER.error("NettyRemotingServer shutdown exception, ", e);
            }
        }
    }


    @Override
    public ChannelEventListener getChannelEventListener() {
        return this.channelEventListener;
    }

    @Override
    public ExecutorService getCallbackExecutor() {
        if (nettyClientConfig.isDisableCallbackExecutor()) {
            return null;
        }
        return callbackExecutor != null ? callbackExecutor : publicExecutor;
    }

    /**
     * @Author jiaxiangen
     * @Description: @方法描述：根据最新的Namesrv地址更新namesrvAddrList地址列表的方法
     * @Date 2025/6/11 13:54
     * @Param [addrs]
     * @Return void
     */
    @Override
    public void updateNameServerAddressList(List<String> addrs) {
        //得到老的客户端
        List<String> old = this.namesrvAddrList.get();
        //定义一个update标识
        boolean update = false;

        if (old == null) {
            update = true;
        } else if (old.size() != addrs.size()) {
            update = true;
        } else {
            for (int i = 0; i < old.size(); i++) {
                if (!old.get(i).equals(addrs.get(i))) {
                    update = true;
                    break;
                }
            }
        }

        if (update) {
            Collections.shuffle(addrs);
            //更新可用的Namesrv地址
            this.namesrvAddrList.set(addrs);
            //当前客户端正在与哪个Namesrv地址进行通信
            //从Namesrv地址选择器中当前正在使用的Namesrv地址，也就是说当前客户端这个Namesrv地址建立了channel连接
            //!addrs.contains(this.namesrvAddrChoosed.get())判断一下，看看当前选择的Namesrv地址是否在最新的Namesrv地址列表中
            if (this.namesrvAddrChoosed.get() != null && !addrs.contains(this.namesrvAddrChoosed.get())) {
                //如果不在的话，说明这个channel不能被使用了，这时候就先得到当前使用的的Namesrv地址，然后关闭这个地址对应的channel
                String namesrvAddr = this.namesrvAddrChoosed.get();
                for (String addr : this.channelTables.keySet()) {
                    //判断当前使用的地址是否在channelTables中
                    if (addr.contains(namesrvAddr)) {
                        //如果存在就得到对应的channel包装器对象
                        ChannelWrapper channelWrapper = this.channelTables.get(addr);
                        if (channelWrapper != null) {
                            //包装器对象不为空则直接关闭这个channel
                            channelWrapper.close();
                        }
                    }
                }
            }
        }


    }

    /**
     * @Author jiaxiangen
     * 在Broker模块启动之后，Broker节点要把自己的信息注册到Namesrv上，如果Namesrv是集群部署的，也就是有多个Namesrv节点，那么Broker节点就要和多个Namesrv节点建立连接
     * 也就是说，在Broker把信息注册到Namesrv上的时候，会循环遍历所有的Namesrv节点的地址，和每一个Namesrv节点都建立channel，把自己的信息注册到每一个Namesrv上
     * 这是在第五版本代码中，调用当前方法的情况，也就是说在调用当前方法的时候，方法参数addr传递的就是Namesrv节点的地址，这一点想必是很容易理解的
     * 但是在当前方法中大家又会看到首先对addr做了一个判空处理，如果addr为空，则调用getAndCreateNameserverChannel()方法，使用getAndCreateNameserverChannel()方法得到一个随机channel
     * 也就是得到一个随机的Namesrv节点的channel，这是为什么呢？那当前方法什么时候传入的方法参数为null呢？在第六版本的代码中会引入一个MQClientAPIImpl类，在这个类中我会添加详细注释为大家解释当前的问题
     * @Date 2025/6/11 15:11
     * @Param
     * @Return
     */
    private Channel getAndCreateChannel(String addr) throws InterruptedException {
        if (addr == null) {
            //如果addr为空，则调用getAndCreateNameserverChannel()方法，使用getAndCreateNameserverChannel()方法得到一个随机channel
            return getAndCreateNameserverChannel();
        }
        //如果addr不为空，则从channelTables中获取对应的channel包装器对象
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        //如果channel包装器对象不存在或者状态不正常，那就创建一个新的channel
        return this.createChannel(addr);
    }

    private Channel createChannel(String addr) throws InterruptedException {
        //从channelTables中获取对应的channel包装器对象
        ChannelWrapper cw = this.channelTables.get(addr);
        //如果channel包装器对象不存在或者状态不正常，那就创建一个新的channel
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        if (lockChannelTables.tryLock(nettyClientConfig.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)) {
            try {

                boolean createNewConnection = false;
                if (cw != null) {
                    if (cw.isOK()) {
                        return cw.getChannel();
                    } else if (!cw.getChannelFuture().isDone()) {
                        //如果channelFuture已经完成，则正在创建连接
                        createNewConnection = false;
                    } else {
                        //当前连接不存在 或者失效
                        this.channelTables.remove(addr);
                        createNewConnection = true;
                    }
                } else {
                    //当前连接不存在
                    createNewConnection = true;
                }
                if (createNewConnection) {
                    String[] hostAndPort = getHostAndPort(addr);
                    ChannelFuture channelFuture = fetchBootstrap(addr)
                            .connect(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
                    cw = new ChannelWrapper(addr, channelFuture);
                    this.channelTables.put(addr, cw);
                    //根据channel获取对应的channel包装器对象
                    this.channelWrapperTables.put(cw.getChannel(), cw);
                }
            } catch (Exception e) {
                LOGGER.error("create channel exception", e);
            } finally {
                lockChannelTables.unlock();
            }
        }
        //cw 不为空说明当前channel包装器对象不为空
        if (cw != null) {
            //返回对应的channel
            return waitChannelFuture(addr, cw);
        }
        return null;
    }

    /**
     * @方法描述：等待Netty客户端和服务端连接完成的方法
     */
    private Channel waitChannelFuture(String addr, ChannelWrapper cw) {
        //从Channel包装器对象中得到ChannelFuture对象
        ChannelFuture channelFuture = cw.getChannelFuture();
        //等待连接操作完成，这里和netty的内容耦合比较深，这里就不展开讲解了，大家可以回顾回顾netty小册的第十四篇文章
        if (channelFuture.awaitUninterruptibly(this.nettyClientConfig.getConnectTimeoutMillis())) {
            if (cw.isOK()) {
                LOGGER.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                //连接完成，并且channel状态是正常的，那就直接返回channel
                return cw.getChannel();
            }
            //连接完成，但是channel状态不正常，记录警告日志
            else {
                LOGGER.warn("createChannel: connect remote host[{}] failed, {}", addr, channelFuture.toString());
            }
        }
        //连接超时，记录警告日志
        else {
            LOGGER.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.nettyClientConfig.getConnectTimeoutMillis(), channelFuture.toString());
        }
        //连接失败，返回null
        return null;
    }

    /**
     * @方法描述：和指定的网络地址创建channel连接的方法
     */
    private Channel getAndCreateNameserverChannel() throws InterruptedException {
        String addr = this.namesrvAddrChoosed.get();
        if (addr != null) {
            ChannelWrapper cw = this.channelTables.get(addr);
            if (cw != null && cw.isOK()) {
                //如果channelWrapper对象不为空，并且状态正常，则直接返回channel
                return cw.getChannel();
            }
        }
        //程序执行到这里意味着当前选择的Namesrv地址为空，或者并没有创建与该地址对应的channel
        //得到存放Namesrv地址列表的集合
        List<String> namesrvList = this.namesrvAddrList.get();
        if (namesrvChannelLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
            try {
                //双层检测锁
                addr = this.namesrvAddrChoosed.get();
                if (addr != null) {
                    ChannelWrapper cw = this.channelTables.get(addr);
                    if (cw != null && cw.isOK()) {
                        //如果channelWrapper对象不为空，并且状态正常，则直接返回channel
                        return cw.getChannel();
                    }
                }
                if (namesrvList != null && !namesrvList.isEmpty()) {
                    //随机选出一个可用的地址
                    int index = this.namesrvAddrIndex.get();
                    index = Math.abs(index) % namesrvList.size();
                    addr = namesrvList.get(index);
                    this.namesrvAddrChoosed.set(addr);
                    //创建channel
                    return createChannel(addr);
                }
                //如果所有地址都尝试了，但是都创建连接失败了，直接抛出连接异常
                throw new RemotingConnectException(namesrvList.toString());
            } catch (Exception e) {
                LOGGER.error("getAndCreateNameserverChannel: create name server channel exception", e);
            } finally {
                namesrvChannelLock.unlock();
            }

        } else {
            LOGGER.warn("getAndCreateNameserverChannel: try to lock name server, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
        }
        return null;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：中断拉取请求操作的方法，该方法在第五版本代码中用不到，大家暂时忽略即可
     */
    private void interruptPullRequests(Set<String> brokerAddrSet) {
        for (ResponseFuture responseFuture : responseTable.values()) {
            RemotingCommand cmd = responseFuture.getRequestCommand();
            if (cmd == null) {
                continue;
            }
            String remoteAddr = RemotingHelper.parseChannelRemoteAddr(responseFuture.getChannel());
            if (brokerAddrSet.contains(remoteAddr) && (cmd.getCode() == 11 || cmd.getCode() == 361)) {
                LOGGER.info("interrupt {}", cmd);
                responseFuture.interrupt();
            }
        }
    }

    @Override
    public List<String> getNameServerAddressList() {

        return this.namesrvAddrList.get();
    }


    @Override
    public List<String> getAvailableNameSrvList() {
        return new ArrayList<>(this.availableNamesrvAddrMap.keySet());
    }

    /*
     * @Author jiaxiangen
     * @方法描述：客户端对外发送请求的方法
     * @Date 2025/6/11 17:17
     * @Param [addr, request, timeoutMillis]
     * @Return java.util.concurrent.CompletableFuture<org.apache.rocketmq.remoting.protocol.RemotingCommand>
     */
    @Override
    public CompletableFuture<RemotingCommand> invoke(String addr, RemotingCommand request, long timeoutMillis) {
        CompletableFuture<RemotingCommand> completableFuture = new CompletableFuture<>();
        try {
            Channel channel = this.getAndCreateChannel(addr);
            if (channel != null && channel.isActive()) {
                return invokeImpl(channel, request, timeoutMillis)
                        .whenComplete((remotingCommand, throwable) -> {
                            if (throwable == null) {
                                updateChannelLastResponseTime(addr);
                            }
                        })
                        .thenApply(ResponseFuture::getRequestCommand);
            } else {
                LOGGER.warn("invoke: create channel failed, {}", addr);
                this.closeChannel(channel);
                completableFuture.completeExceptionally(new RemotingConnectException(addr));
            }
        } catch (Exception e) {
            LOGGER.error("invoke: invoke remote exception, {}", e);
            completableFuture.completeExceptionally(e);
        }
        return completableFuture;
    }


    @Override
    public CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
                                                        final long timeoutMillis) {
        //得到操作的开始时间
        Stopwatch stopwatch = Stopwatch.createStarted();
        //调用父类方法，向服务端发送请求
        return super.invokeImpl(channel, request, timeoutMillis).thenCompose(responseFuture -> {
            //接收到响应之后，在这里获得响应对象
            RemotingCommand response = responseFuture.getResponseCommand();
            //如果响应对象的响应码是GO_AWAY，在之前构建Netty服务端的时候大家应该见过，如果服务端正在执行关闭操作，这个时候接收到客户端请求了
            //就会回复给客户端一个GO_AWAY响应码，意味着服务端正在关闭
            if (response.getCode() == ResponseCode.GO_AWAY) {
                //如果netty客户端配置了接收到GO_AWAY响应码之后，允许重连的操作，那接下来就执行重连操作
                if (nettyClientConfig.isEnableReconnectForGoAway()) {
                    //从channelWrapperTables中获取与channel对应的ChannelWrapper包装器对象
                    ChannelWrapper channelWrapper = channelWrapperTables.computeIfPresent(channel, (channel0, channelWrapper0) -> {
                        try {
                            //如果channelWrapperTables中存在与channel对应的ChannelWrapper包装器对象，那就执行重连操作
                            if (channelWrapper0.reconnect()) {
                                LOGGER.info("Receive go away from channel {}, recreate the channel", channel0);
                                //如果重连成功，那就把新的ChannelWrapper包装器对象放入channelWrapperTables中
                                channelWrapperTables.put(channelWrapper0.getChannel(), channelWrapper0);
                            }
                        } catch (Throwable t) {
                            LOGGER.error("Channel {} reconnect error", channelWrapper0, t);
                        }
                        return channelWrapper0;
                    });
                    //以下是执行请求重试的操作，注意，这个时候判断了channelWrapper是否为null，如果不为null，这个时候就意味着已经执行了重连操作
                    //channel已经是新的了
                    if (channelWrapper != null) {
                        //判断是否配置了允许请求重试
                        if (nettyClientConfig.isEnableTransparentRetry()) {
                            //计算已经耗费的时间
                            long duration = stopwatch.elapsed(TimeUnit.MILLISECONDS);
                            stopwatch.stop();
                            //创建新的请求对象
                            RemotingCommand retryRequest = RemotingCommand.createRequestCommand(request.getCode(), request.readCustomHeader());
                            retryRequest.setBody(request.getBody());
                            Channel retryChannel;
                            //判断channel状态是否正常
                            if (channelWrapper.isOK()) {
                                //状态正常则得到新创建的channel
                                retryChannel = channelWrapper.getChannel();
                            } else {
                                //状态不正常则等待重连操作完成，得到新创建的channel
                                retryChannel = waitChannelFuture(channelWrapper.getChannelAddress(), channelWrapper);
                            }
                            //如果重连成功，那么重连的channel肯定和原来的channel不一样了
                            if (retryChannel != null && channel != retryChannel) {
                                //这个时候就可以调用父类方法，向服务端发送请求了
                                return super.invokeImpl(retryChannel, retryRequest, timeoutMillis - duration);
                            }
                        }
                    }
                }
            }
            return CompletableFuture.completedFuture(responseFuture);
        });
    }

    @Override
    public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
            throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
        //得到开始执行操作的时间
        long beginStartTime = System.currentTimeMillis();
        //得到发送请求的channel
        final Channel channel = this.getAndCreateChannel(addr);
        //解析channel得到服务器的远程地址
        String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
        //判断通道是否存在并且是活跃的
        if (channel != null && channel.isActive()) {
            //得到发送请求的超时时间
            long left = timeoutMillis;
            try {
                //得到到目前为止已经花费的时间，现在还没有发送请求呢
                long costTime = System.currentTimeMillis() - beginStartTime;
                //得到剩余的请求超时时间
                left -= costTime;
                //如果请求超时时间小于零，意味着已经不能再发送请求了，还没发送呢都已经超过超时时间了，直接抛出超时异常即可
                if (left <= 0) {
                    throw new RemotingTimeoutException("invokeSync call the addr[" + channelRemoteAddr + "] timeout");
                }
                //在这里把请求真正发送出去了，这里是同步发送请求的操作，得到服务端回复的响应
                RemotingCommand response = this.invokeSyncImpl(channel, request, left);
                //在这里把channel最新收到响应的时间更新一下
                updateChannelLastResponseTime(addr);
                return response;
            } catch (RemotingSendRequestException e) {
                LOGGER.warn("invokeSync: send request exception, so close the channel[{}]", channelRemoteAddr);
                //出现异常则直接关闭channel
                this.closeChannel(addr, channel);
                throw e;
            } catch (RemotingTimeoutException e) {
                //捕获超时异常之后，判断一下是否真的需要关闭channel，判断一下发送请求的剩余时间是否达到阈值了
                boolean shouldClose = left > MIN_CLOSE_TIMEOUT_MILLIS || left > timeoutMillis / 4;
                //如果达到阈值了，并配配置信息中配置了客户端要关闭超时的socket，那就执行关闭channel的操作
                if (nettyClientConfig.isClientCloseSocketIfTimeout() && shouldClose) {
                    this.closeChannel(addr, channel);
                    LOGGER.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, channelRemoteAddr);
                }
                LOGGER.warn("invokeSync: wait response timeout exception, the channel[{}]", channelRemoteAddr);
                throw e;
            }
        } else {
            //程序执行到这里意味着channel不存在或者不是活跃的，直接抛出连接异常即可
            this.closeChannel(addr, channel);
            throw new RemotingConnectException(addr);
        }
    }


    @Override
    public void invokeAsync(String channelRemoteAddr, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback) throws InterruptedException, RemotingConnectException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        //开始执行时间
        long startTime = System.currentTimeMillis();
        final Channel channel = this.getAndCreateChannel(channelRemoteAddr);
        //判断channel是否可用
        if (channel != null && channel.isActive()) {
            long cost = System.currentTimeMillis() - startTime;
            if (timeoutMillis > cost) {
                //这里抛出的异常从名字上来看，应该是请求过多异常，但是这里怎么会和请求太多有关系呢？我想不明白
                //在得到开始时间和计算消耗时间中间就 final Channel channel = this.getAndCreateChannel(addr);这一行代码执行时会获得同步锁，也可能会创建新的channel
                //这些操作可能会耗费一定的时间，如果这个时间超过了用户定义的超时时间，那就直接抛出超时异常即可，但这里抛出了请求太多异常，我不太明白，有明白的朋友可以帮我解惑一下
                throw new RemotingTooMuchRequestException("invokeAsync call the addr[" + channelRemoteAddr + "] timeout");
            }
            invokeAsyncImpl(channel, request, timeoutMillis - cost, new InvokeCallbackWapper(invokeCallback, channelRemoteAddr));
        } else {
            //执行到这里意味着channel不存在或者不是活跃的，直接抛出连接异常即可
            this.closeChannel(channelRemoteAddr, channel);
            throw new RemotingConnectException(channelRemoteAddr);
        }
    }

    @Override
    public void invokeOneway(String addr, RemotingCommand request, long timeoutMillis) throws InterruptedException, RemotingConnectException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        final Channel channel = this.getAndCreateChannel(addr);
        String remoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
        if (channel != null && channel.isActive()) {
            //执行请求前，执行rpc钩子
            doBeforeRpcHooks(remoteAddr, request);
            //发送请求
            this.invokeOnewayImpl(channel, request, timeoutMillis);
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingConnectException(addr);
        }
    }

    @Override
    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        ExecutorService executorThis = executor;
        if (null == executor) {
            executorThis = this.publicExecutor;
        }
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
        this.processorTable.put(requestCode, pair);
    }


    @Override
    public void setCallbackExecutor(final ExecutorService callbackExecutor) {
        this.callbackExecutor = callbackExecutor;
    }


    @Override
    public boolean isChannelWritable(String addr) {
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.isWritable();
        }
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：判断指定地址是否可用的方法
     */
    @Override
    public boolean isAddressReachable(String addr) {
        if (addr == null || addr.isEmpty()) {
            return false;
        }
        try {
            //为指定地址构建channel连接
            Channel channel = getAndCreateChannel(addr);
            //判断构建完毕的channel的状态是否活跃
            return channel != null && channel.isActive();
        } catch (Exception e) {
            LOGGER.warn("Get and create channel of {} failed", addr, e);
            return false;
        }
    }

    @Override
    public void closeChannels(List<String> addrList) {
        for (String addr : addrList) {
            ChannelWrapper cw = this.channelTables.get(addr);
            if (cw == null) {
                continue;
            }
            this.closeChannel(addr, cw.getChannel());
        }
        //中断拉取请求的操作
        interruptPullRequests(new HashSet<>(addrList));
    }

    public Bootstrap fetchBootstrap(String channelAddress) {
        return this.bootstrap;
    }

    public String[] getHostAndPort(String address) {
        int split = address.lastIndexOf(":");
        return split < 0 ? new String[]{address} : new String[]{address.substring(0, split), address.substring(split + 1)};
    }

    public void closeChannel(final String addr, final Channel channel) {
        if (null == channel) {
            return;
        }
        String addrRemote = null == addr ? RemotingHelper.parseChannelRemoteAddr(channel) : addr;
        try {
            if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    //定义一个是否要从channelTables中移指定channel的标志
                    boolean removeItemFromTable = true;
                    ChannelWrapper prevCW = this.channelTables.get(addrRemote);

                    if (null == prevCW) {
                        removeItemFromTable = false;
                        LOGGER.info("closeChannel: the channel[{}] has been removed from channel table before", addrRemote);
                        return;
                    } else if (prevCW.getChannel() != channel) {
                        LOGGER.info("closeChannel: the channel[{}] has been removed from channel table before, and has been created again, nothing to do.", addrRemote);
                        removeItemFromTable = false;
                    }
                    if (removeItemFromTable) {
                        ChannelWrapper channelWrapper = this.channelWrapperTables.remove(channel);
                        if (channelWrapper != null && channelWrapper.tryClose(channel)) {
                            //从channelTables中移除指定的channel
                            this.channelTables.remove(channel);
                        }
                        LOGGER.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                    }

                    RemotingHelper.closeChannel(channel);


                } catch (Exception e) {
                    LOGGER.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            }
        } catch (Exception e) {
            LOGGER.error("closeChannel exception", e);
        }
    }

    private void updateChannelLastResponseTime(final String addr) {
        //如果地址为空，则从当前选择的Namesrv地址中获取地址
        String address = addr;
        if (address == null) {
            address = this.namesrvAddrChoosed.get();
        }
        //如果地址还是为空，则记录告警日志，直接退出当前方法
        if (address == null) {
            LOGGER.warn("[updateChannelLastResponseTime] could not find address!!");
            return;
        }
        //根据地址从channelTables中获取对应的channel包装器对象
        ChannelWrapper channelWrapper = this.channelTables.get(address);
        //更新最新的响应时间
        if (channelWrapper != null && channelWrapper.isOK()) {
            channelWrapper.updateLastResponseTime();
        }
    }

    /**
     * @Author jiaxiangen
     * @方法描述：关闭指定channel的方法，这个方法和上面的方法逻辑基本一致，只不过这个方法并没有直接根据服务器网络地址来关闭channel
     * @Date 2025/6/11 15:58
     * @Param [channel]
     * @Return void
     */
    public void closeChannel(final Channel channel) {
        if (null == channel) {
            return;
        }
        try {
            if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    ChannelWrapper prevCW = null;
                    String addrRemote = null;
                    //在这里遍历channelTables集合，找到要关闭的channel
                    for (Map.Entry<String, ChannelWrapper> entry : channelTables.entrySet()) {
                        //得到当前遍历的channel的网络地址
                        String key = entry.getKey();
                        //得到当前遍历的channel的包装器对象
                        ChannelWrapper prev = entry.getValue();
                        //从channel包装器对象中得到channel对象
                        if (prev.getChannel() != null) {
                            //判断当前遍历的channel是否和用户指定要关闭的channel一致
                            if (prev.getChannel() == channel) {
                                //如果一直则得到当前遍历的channel的包装器对象
                                prevCW = prev;
                                //得到当前遍历的channel的网络地址
                                addrRemote = key;
                                //打断循环
                                break;
                            }
                        }
                    }
                    //如果没有找到对应的channel包装器对象，则更新不需要移除channel的标志
                    if (null == prevCW) {
                        LOGGER.info("eventCloseChannel: the channel[{}] has been removed from the channel table before", addrRemote);
                        removeItemFromTable = false;
                    }
                    if (removeItemFromTable) {
                        ChannelWrapper channelWrapper = this.channelWrapperTables.remove(channel);
                        if (channelWrapper != null && channelWrapper.tryClose(channel)) {
                            this.channelTables.remove(addrRemote);
                        }
                        LOGGER.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                        RemotingHelper.closeChannel(channel);
                    }
                } catch (Exception e) {
                    LOGGER.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                LOGGER.warn("closeChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
            }
        } catch (InterruptedException e) {
            LOGGER.error("closeChannel exception", e);
        }
    }


    class ChannelWrapper {
        private final ReentrantReadWriteLock lock;

        private ChannelFuture channelFuture;

        private ChannelFuture channelToClose;

        private long lastResponseTime;

        private volatile long lastReconnectTime;


        private final String channelAddress;

        public ChannelWrapper(String channelAddress, ChannelFuture channelFuture) {
            this.lock = new ReentrantReadWriteLock();
            this.channelAddress = channelAddress;
            this.channelFuture = channelFuture;
            this.lastResponseTime = System.currentTimeMillis();
        }

        public boolean isOK() {
            Channel channel = getChannel();
            if (channel != null && channel.isActive()) {
                return true;
            }
            return false;
        }

        public void updateLastResponseTime() {
            this.lastResponseTime = System.currentTimeMillis();
        }

        public boolean reconnect() {
            lock.writeLock().lock();
            try {
                if (lastReconnectTime == 0 || System.currentTimeMillis() - this.lastReconnectTime > Duration.ofSeconds(nettyClientConfig.getMaxReconnectIntervalTimeSeconds()).toMillis()) {
                    channelToClose = channelFuture;
                    channelFuture = fetchBootstrap(channelAddress).
                            connect(new InetSocketAddress(getHostAndPort(channelAddress)[0],
                                    Integer.parseInt(getHostAndPort(channelAddress)[1])));
                    lastReconnectTime = System.currentTimeMillis();
                    return true;
                }

            } finally {
                lock.writeLock().unlock();
            }
            return false;
        }

        //尝试关闭指定channel的方法
        public boolean tryClose(Channel channel) {
            try {
                lock.readLock().lock();
                if (channelFuture != null) {
                    if (channelFuture.channel().equals(channel)) {
                        return true;
                    }
                }
            } finally {
                lock.readLock().unlock();
            }
            return false;
        }

        public void close() {
            try {
                lock.writeLock().lock();
                if (channelFuture != null) {
                    closeChannel(channelFuture.channel());
                }
                if (channelToClose != null) {
                    closeChannel(channelToClose.channel());
                }
            } finally {
                lock.writeLock().unlock();
            }
        }

        private Channel getChannel() {
            return getChannelFuture().channel();
        }

        public ReentrantReadWriteLock getLock() {
            return lock;
        }

        public ChannelFuture getChannelFuture() {
            lock.readLock().lock();
            try {
                return this.channelFuture;
            } finally {
                lock.readLock().unlock();
            }
        }

        public void setChannelFuture(ChannelFuture channelFuture) {
            this.channelFuture = channelFuture;
        }

        public ChannelFuture getChannelToClose() {
            return channelToClose;
        }

        public void setChannelToClose(ChannelFuture channelToClose) {
            this.channelToClose = channelToClose;
        }

        public long getLastResponseTime() {
            return lastResponseTime;
        }

        public void setLastResponseTime(long lastResponseTime) {
            this.lastResponseTime = lastResponseTime;
        }

        public long getLastReconnectTime() {
            return lastReconnectTime;
        }

        public void setLastReconnectTime(long lastReconnectTime) {
            this.lastReconnectTime = lastReconnectTime;
        }

        public String getChannelAddress() {
            return channelAddress;
        }

        //判断channel是否可写的方法
        public boolean isWritable() {
            return getChannel().isWritable();
        }
    }

    //Netty客户端的事件处理器，专门处理客户端接收到的消息
    class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> {


        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
            processMessageReceived(ctx, msg);
        }
    }


    //Netty客户端的连接管理处理器，专门处理客户端连接的事件，这个连接管理器在构建Netty服务端的时候已经见过了，所以我就不添加详细注释了
    class NettyConnectManageHandler extends ChannelDuplexHandler {


        //连接建立成功时该方法会被回调
        @Override
        public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
            final String local = localAddress == null ? "UNKNOWN" : RemotingHelper.parseSocketAddressAddr(localAddress);
            final String remote = remoteAddress == null ? "UNKNOWN" : RemotingHelper.parseSocketAddressAddr(remoteAddress);
            LOGGER.info("NETTY CLIENT PIPELINE: CONNECT  {} => {}", local, remote);
            super.connect(ctx, remoteAddress, localAddress, promise);
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remote, ctx.channel()));
            }
        }


        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            LOGGER.info("NETTY CLIENT PIPELINE: ACTIVE, {}", remoteAddress);
            super.channelActive(ctx);
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.ACTIVE, remoteAddress, ctx.channel()));
            }
        }


        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            LOGGER.info("NETTY CLIENT PIPELINE: DISCONNECT {}", remoteAddress);
            closeChannel(ctx.channel());
            super.disconnect(ctx, promise);
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
            }
        }


        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            LOGGER.info("NETTY CLIENT PIPELINE: CLOSE {}", remoteAddress);
            closeChannel(ctx.channel());
            super.close(ctx, promise);
            NettyRemotingClient.this.failFast(ctx.channel());
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
            }
        }


        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            LOGGER.info("NETTY CLIENT PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            closeChannel(ctx.channel());
            super.channelInactive(ctx);
        }


        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    LOGGER.warn("NETTY CLIENT PIPELINE: IDLE exception [{}]", remoteAddress);
                    closeChannel(ctx.channel());
                    if (NettyRemotingClient.this.channelEventListener != null) {
                        NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
                    }
                }
            }
            ctx.fireUserEventTriggered(evt);
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            LOGGER.warn("NETTY CLIENT PIPELINE: exceptionCaught {}", remoteAddress);
            LOGGER.warn("NETTY CLIENT PIPELINE: exceptionCaught exception.", cause);
            closeChannel(ctx.channel());
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
            }
        }
    }

    class InvokeCallbackWapper implements InvokeCallback {
        private final InvokeCallback invokeCallback;

        private final String addr;

        public InvokeCallbackWapper(InvokeCallback invokeCallback, String addr) {

            this.invokeCallback = invokeCallback;
            this.addr = addr;
        }

        @Override
        public void operationComplete(ResponseFuture responseFuture) {
            this.invokeCallback.operationComplete(responseFuture);
        }

        @Override
        public void operationSucceed(RemotingCommand remotingCommand) {
            updateChannelLastResponseTime(addr);
            this.invokeCallback.operationSucceed(remotingCommand);
        }

        @Override
        public void operationFail(final Throwable throwable) {
            this.invokeCallback.operationFail(throwable);
        }
    }
}
